package com.azzcs.springbootkafka.comtroller;

import com.azzcs.springbootkafka.service.KafkaService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

/**
 * @Author: wzg
 * @Date: 2021/4/15 上午10:37
 */
@Slf4j
@RestController
public class KafkaController {

    @Autowired
    private KafkaTemplate<Integer,String> kafkaTemplate;

    @Autowired
    public KafkaService kafkaService;

    @GetMapping("/send/message/{message}")
    public String sendMessage(@PathVariable("message")String message) throws ExecutionException, InterruptedException {
        kafkaService.sendSyncMessage(message);
        return "ok";
    }

    @GetMapping("/asyn/send/message/{message}")
    public String asynSendMessage(@PathVariable("message")String message) throws ExecutionException, InterruptedException {
        ListenableFuture<SendResult<Integer, String>> feature = kafkaTemplate.send("spring-boot-topic", 0, 0, message);
        feature.addCallback(new ListenableFutureCallback<SendResult<Integer,String>>(){

            @Override
            public void onSuccess(SendResult<Integer, String> result) {
                RecordMetadata metadata = result.getRecordMetadata();
                log.info(metadata.topic());
                log.info(""+metadata.partition());
                log.info(""+metadata.offset());
            }

            @Override
            public void onFailure(Throwable throwable) {
                log.error("",throwable);
            }
        });
        return "ok";
    }

    @GetMapping("tx/send/message/{message}")
    public String txSendMessage(@PathVariable("message") String message){
        kafkaService.sendMessage(message);
        return "ok";
    }
}
